package com.webchat.pay.service;


import com.webchat.common.bean.APIResponseBean;
import com.webchat.common.constants.WebConstant;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.service.SnowflakeIdGeneratorService;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.util.AESEncryptUtil;
import com.webchat.common.util.DateUtils;
import com.webchat.common.util.HttpClientUtil;
import com.webchat.common.util.JsonUtil;
import com.webchat.common.util.ThreadPoolExecutorUtil;
import com.webchat.domain.dto.payment.CallBackRequestBodyDTO;
import com.webchat.domain.dto.payment.PaymentCallbackEncryptBody;
import com.webchat.domain.dto.payment.PaymentResultDTO;
import com.webchat.domain.dto.queue.ArticleDelayMessageDTO;
import com.webchat.domain.dto.queue.PaymentCallBackQueueDTO;
import com.webchat.domain.vo.response.payment.AppBaseResponseVO;
import com.webchat.pay.messagequeue.PaymentCallBackRedisDelayQueue;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 支付结果回调通知服务
 *
 *
 */
@Service
public class PaymentCallBackService {


    @Autowired
    private PaymentAppService paymentAppService;

    @Autowired
    private SnowflakeIdGeneratorService snowflakeIdGeneratorService;

    @Autowired
    private MessageQueueProducer<PaymentCallBackQueueDTO, Long> callBackQueueProducer;

    @Autowired
    @Qualifier(value = "paymentCallBackExecutor")
    private Executor paymentCallBackExecutor;

    private static final int MAX_LOCAL_RETRY_TIMES = 2;

    private static final int CALLBACK_FIRST_TIME = 1;

    /**
     * 2.1 首次回调：实时通知我们接入方支付结果
     *
     * 2.2 第一次回调失败，我们重试机制（失败原因：网络波动、对方服务异常）
     * 2.2 重拾试策略：实时重试（第2次0s后）、（1min后）、（10min后）、（1h后）、（24h后）、人工介入
     * 3. 回调成功如何判断？ ACK机制，需要接入方收到回调后响应我们回调处理结果（true）
     * 4. 重试策略我们如何设计？
     * 4.1 调度（回调任务任务状态、下一次回调时间）
     * 4.2 延迟队列（最佳实践）
     */
    public void callBack(PaymentResultDTO paymentResult) {

        // 异步处理回调，不足则支付主流程
        ThreadPoolExecutorUtil.execute(() -> doCallBack(paymentResult));
    }

    public void doCallBack(PaymentResultDTO paymentResult) {

        // 第一次回调
        AtomicInteger callBackTimes = new AtomicInteger(CALLBACK_FIRST_TIME);
        boolean callBackResult = this.doExecCallBack(paymentResult, callBackTimes);
        if (callBackResult) {
            // TODO
            return;
        }
        int nextRetryTimes = MAX_LOCAL_RETRY_TIMES + 1;
        Long nextTime = this.getNextCallBackTime(nextRetryTimes);
        // 雪花算法生成taskId
        String taskId = snowflakeIdGeneratorService.generateId();
        // 任务消息入队
        this.push2DelayQueue(taskId, paymentResult, nextRetryTimes, nextTime);
    }

    public void doBatchCallBack(List<PaymentCallBackQueueDTO> queueMessages) {

        List<CompletableFuture<Boolean>> futures = new ArrayList<>();

        /**
         * 基于自定义线程池paymentCallBackExecutor，并行出执行批量回调任务
         *
         */
        for (PaymentCallBackQueueDTO paymentCallBackQueueDTO : queueMessages) {
            PaymentResultDTO paymentResult = paymentCallBackQueueDTO.getData();
            CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() ->
                    doExecCallBack(paymentResult, new AtomicInteger(CALLBACK_FIRST_TIME)), paymentCallBackExecutor);
            futures.add(future);
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        for (int i = 0; i < futures.size(); i++) {
            PaymentCallBackQueueDTO callBackQueue = queueMessages.get(i);
            try {
                Boolean callBackRes = futures.get(i).get();
                if (callBackRes) {
                    // TODO
                    continue;
                }
            } catch (Exception e) {

            }
            int nextRetryTimes = callBackQueue.getRetryTimes() + 1;
            Long nextTime = this.getNextCallBackTime(nextRetryTimes);
            if (nextTime == null) {
                // 超过最大重试次数
                // TODO
                continue;
            }
            this.push2DelayQueue(callBackQueue.getTaskId(),
                                 callBackQueue.getData(),
                                 nextRetryTimes,
                                 nextTime);
        }
    }

    /**
     * 回调
     * @return
     */
    private boolean doExecCallBack(PaymentResultDTO paymentResult, AtomicInteger callBackTimes) {

        if (callBackTimes.getAndAdd(1) > MAX_LOCAL_RETRY_TIMES ) {
            return false;
        }
        // 首先获取接入方回调接口配置
        CallBackRequestBodyDTO data = paymentResult.getData();
        AppBaseResponseVO app = paymentAppService.appInfo(data.getAppId());
        String callBack = app.getCallBack();
        String encryptKey = app.getEncryptKey();
        if (StringUtils.isBlank(callBack) || StringUtils.isBlank(encryptKey)) {
            return true;
        }
        String callbackJsonMessage = JsonUtil.toJsonString(paymentResult);
        String encryptMessage = AESEncryptUtil.encrypt(callbackJsonMessage, encryptKey);
        PaymentCallbackEncryptBody callbackMessage = new PaymentCallbackEncryptBody(encryptMessage);
        try {
            APIResponseBean apiResponseBean = HttpClientUtil.postObjectForUrl(callBack, callbackMessage, APIResponseBean.class);
            // 收到接入方ACK
            return apiResponseBean.getCode() == WebConstant.HTTP_STATUS_OK;
        } catch (Exception e) {
            // TODO
            doExecCallBack(paymentResult, callBackTimes);
        }
        return false;
    }

    /**
     * 入队
     *
     * @param taskId
     * @param data
     * @param retryTimes
     * @param time
     */
    private void push2DelayQueue(String taskId,
                                 PaymentResultDTO data,
                                 int retryTimes,
                                 Long time) {

        PaymentCallBackQueueDTO queueMessage = PaymentCallBackQueueDTO.of(taskId,
                data,
                retryTimes,
                time);
        callBackQueueProducer.prioritySend(MessageQueueEnum.QUEUE_PAYMENT_CALL_BACK_TASK_MESSAGE,
                queueMessage, time);
    }


    /**
     * 根据当前重试次数，计算任务下一次时间时刻
     *
     * @param nextRetryTimes
     * @return
     */
    private Long getNextCallBackTime(int nextRetryTimes) {
        // 计算队列当前重试次数，需要减掉前两次的回调主线程的重试次数
        int queueCurrRetryTimes = nextRetryTimes - MAX_LOCAL_RETRY_TIMES;
        if (WebConstant.PAYMENT_CALL_BACK_TIMES.length <= queueCurrRetryTimes) {
            return null;
        }
        long nextTime = WebConstant.PAYMENT_CALL_BACK_TIMES[queueCurrRetryTimes - 1];
        return DateUtils.getCurrentTimeMillis() + nextTime;
    }
}
